package main.scala.demo.flume

import org.apache.flume.Event
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.flume.FlumeUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * MyFlumeStreamDemo
  *
  * @author zhangyimin
  * @date 2018-10-17 下午5:27
  * @version 1.0
  */
object MyFlumePushStreamDemo {
  def main(args: Array[String]): Unit = {
    //创建SparkStreaming对象
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)
    //local[2]代表核数为2
    //    val spark = SparkSession.builder().appName("MyNetworkWordCount").master("local[2]").getOrCreate()
    //    val sc = spark.sparkContext
    val sparkConf = new SparkConf().setAppName("MyNetworkWordCount").setMaster("local[2]")
    val streamingContext = new StreamingContext(sparkConf, Seconds(3))
    //创建一个DStream
    val flumeDStream = FlumeUtils.createStream(streamingContext, "10.16.7.36", 1234)
    //flume主动推送数据给sparkStreaming e表示event flume中传输数据的基本单位
      val data=flumeDStream.map(e=>({
      new String(e.event.getBody.array())
    }))
    data.print()

    streamingContext.start()
    streamingContext.awaitTermination()
  }
}

object MyFlumePollStreamDemo {
  def main(args: Array[String]): Unit = {
    //创建SparkStreaming对象
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)
    //local[2]代表核数为2
    //    val spark = SparkSession.builder().appName("MyNetworkWordCount").master("local[2]").getOrCreate()
    //    val sc = spark.sparkContext
    val sparkConf = new SparkConf().setAppName("MyNetworkWordCount").setMaster("local[2]")
    val streamingContext = new StreamingContext(sparkConf, Seconds(3))
    //创建一个DStream
    //ip 和 port 为flume的IP和端口
    val flumeDStream = FlumeUtils.createPollingStream(streamingContext, "10.16.7.36", 1234,StorageLevel.MEMORY_ONLY_2)
    //sparkStreaming从flume拉取pull数据 e表示event flume中传输数据的基本单位
    val data=flumeDStream.map(e=>({
      new String(e.event.getBody.array())
    }))
    data.print()

    streamingContext.start()
    streamingContext.awaitTermination()
  }
}


